package com.firmiana.jt809.netty.main;

import com.firmiana.jt809.common.ConnectionStateEnum;
import com.firmiana.jt809.netty.JT809MessageDecoder;
import com.firmiana.jt809.netty.JT809MessageEncoder;
import com.firmiana.jt809.service.AbstractBaseService;
import com.firmiana.jt809.service.Listener;
import com.firmiana.jt809.service.ServiceException;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 809主链路
 * 下级平台向上级平台发送数据
 *
 * @author hucl
 * @date 2020/8/25 14:18
 */
@Slf4j
public class NettyTCPMainLink extends AbstractBaseService {
    protected final AtomicReference<ConnectionStateEnum> serverState = new AtomicReference<>(ConnectionStateEnum.Created);

    protected final int port;
    protected final String host;
    protected EventLoopGroup bossGroup;
    protected EventLoopGroup workerGroup;

    public NettyTCPMainLink(int port) {
        this.port = port;
        this.host = null;
    }

    public NettyTCPMainLink(int port, String host) {
        this.port = port;
        this.host = host;
    }

    @Override
    public void init() {
        if (!serverState.compareAndSet(ConnectionStateEnum.Created, ConnectionStateEnum.Initialized)) {
            throw new ServiceException("Server already init");
        }
    }

    @Override
    public boolean isRunning() {
        return serverState.get() == ConnectionStateEnum.Started;
    }

    @Override
    public void start(Listener listener) {
        if (!serverState.compareAndSet(ConnectionStateEnum.Initialized, ConnectionStateEnum.Starting)) {
            throw new ServiceException("Server already started or have not init");
        }
        createNioServer(listener);
    }

    private void createNioServer(Listener listener) {
        EventLoopGroup bossGroup = getBossGroup();
        EventLoopGroup workerGroup = getWorkerGroup();

        if (bossGroup == null) {
            bossGroup = new NioEventLoopGroup(getBossThreadNum());
        }
        if (workerGroup == null) {
            workerGroup = new NioEventLoopGroup(getWorkerThreadNum());
        }
        createServer(listener, bossGroup, workerGroup);
    }

    private void createServer(Listener listener, EventLoopGroup boss, EventLoopGroup work) {
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss, work)
                    .channel(NioServerSocketChannel.class)
                    // todo 添加 netty自带的心跳机制IdleStateHandler
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            initPipeline(ch.pipeline());
                        }
                    });
            /**
             * 绑定端口并启动去接收进来的连接
             */
            InetSocketAddress address = new InetSocketAddress(host, port);
            bootstrap.bind(address).addListener(future -> {
                if (future.isSuccess()) {
                    serverState.set(ConnectionStateEnum.Started);
                    log.info("main link start success on:{}", port);
                    if (listener != null) listener.onSuccess(port);
                } else {
                    log.error("main link start failure on:", future.cause());
                    if (listener != null) listener.onFailure(future.cause());
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    protected void initPipeline(ChannelPipeline pipeline) {
        // 主链路心跳机制，连续3min没读到下级平台发送的请求，则断开连接
        pipeline.addLast(new IdleStateHandler(3, 0 , 0, TimeUnit.MINUTES));
        pipeline.addLast("decoder", getDecoder());
        pipeline.addLast("encoder", getEncoder());
        pipeline.addLast("handler", getChannelHandler());
    }

    @Override
    public void reconnect() {
        super.reconnect();
    }

    @Override
    public void stop(Listener listener) {
        if (!serverState.compareAndSet(ConnectionStateEnum.Started, ConnectionStateEnum.Shutdown)) {
            if (listener != null) {
                listener.onFailure(new ServiceException("main links was already shutdown."));
            }
            log.error("{} was already shutdown.", this.getClass().getSimpleName());
            return;
        }
        log.info("try shutdown {}...", this.getClass().getSimpleName());
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
        }
        if (workerGroup!=null) {
            workerGroup.shutdownGracefully().syncUninterruptibly();
        }
        log.info("{} shutdown successfully...", this.getClass().getSimpleName());
        if (listener != null) {
            listener.onSuccess(port);
        }
    }

    protected ChannelHandler getDecoder() {
        return new JT809MessageDecoder();
    }

    protected ChannelHandler getEncoder() {
        return new JT809MessageEncoder();
    }

    protected ChannelHandler getChannelHandler() {
        return new MainLinkHandler();
    }

    public EventLoopGroup getBossGroup() {
        return bossGroup;
    }

    public EventLoopGroup getWorkerGroup() {
        return workerGroup;
    }

    protected int getBossThreadNum() {
        return 1;
    }

    protected int getWorkerThreadNum() {
        return 0;
    }

}
